package com.nju.graphx

import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{DriverManager, ResultSet}
import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, ListBuffer}

/**
 *
 * @authorliyunfei
 * @date2022/12/11
 * */
object UserRecommend {

  def createUsersGraph()={

  }

  def main(args: Array[String]): Unit = {
    val sc: SparkContext = GraphxSparkStarter.initSpark()
    // 1.从存储DB获取数据集
    val result: ResultSet = GraphxDataServiceMySql.queryUsers()
    //构建点集和边集
    val userVertexs = new ListBuffer[(Long, String)]()
    val userEdges = ListBuffer[Edge[Int]]()
    //用户与关注用户关系
    val userFuser = mutable.Map[Long, mutable.Set[Long]]()
    while (result.next()) {
      var userid = result.getString("user_id").toLong
      var username = result.getString("user_name")
      userVertexs.append((userid, username))
      var longs: ArrayBuffer[Long] = ArrayBuffer[Long]()
      var sets = mutable.Set[Long]()
      var str = result.getString("follows_id")
      if (str != "") {
        str.split(",").foreach(x => longs.append(x.toLong))
        for (i <- longs) {
          userEdges.append(Edge(userid, i, 1))
          sets.add(i)
        }
      }
      userFuser.put(userid, sets)
    }
    val vertexRDD: RDD[(Long, String)] = sc.makeRDD(userVertexs)
    //    println(vertexRDD.foreach(println))
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(userEdges)
    //    println(edgeRDD.foreach(println))
    // 构建图
    val graph: Graph[String, Int] = Graph(vertexRDD, edgeRDD)

    //recommendUsers推荐结果，key为用户，value为给key用户推荐的用户集合
    val recommendUsers = mutable.Map[Long, mutable.Set[Long]]()
    val k = 3
    for ((userid, _) <- userFuser) {
      // 单个用户推荐(遍历所有用户推荐情况)
      recommendUser(userid, k)
    }

    println("start to storage to mysql")
    storageToMySql(recommendUsers)
    def recommendUser(userid: Long, k: Int): Unit = {
      for (user <- userFuser) {
        if (user._1 != userid) {
          val sets = userFuser.getOrElseUpdate(userid, mutable.Set[Long]())
          val bool: Boolean = user._2.intersect(sets).size >= k

          var useridToUser = mutable.Set[Long]()
          var userToUserid = mutable.Set[Long]()
          if (bool) {
            useridToUser ++= user._2 diff sets
            userToUserid ++= sets diff user._2
          }
          recommendUsers.getOrElseUpdate(userid, mutable.Set[Long]()) ++= userToUserid
          recommendUsers.getOrElseUpdate(user._1, mutable.Set[Long]()) ++= useridToUser
        }
      }
    }
  }

  /**
   * 存入MySQL数据库
   *
   * @param recommendUsers
   */
  private def storageToMySql(recommendUsers:mutable.Map[Long, mutable.Set[Long]]): Unit ={
         // 暴力遍历插入
         recommendUsers.foreach(u => {
            val userId = u._1
            val recommendUserIdSet = u._2
            val ids: String = recommendUserIdSet.mkString(",")
            println(ids)
            val argList: ListBuffer[Any] = new ListBuffer[Any]()
            argList += userId
            argList += ids
            GraphxDataServiceMySql.storageUserRecommend(argList)
    })
  }

}
